package realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import realtime.util.DateFormatUtil;
import realtime.util.MyKafkaUtil;

public class BaseLogApp {
    public static void main(String[] args) throws Exception {

        //配置
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        /*env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flinkCDC");
        env.setStateBackend(new HashMapStateBackend());*/

        System.setProperty("HADOOP_USER_NAME", "root");


        //从kafka拉数据
        DataStreamSource<String> kafkaStream = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("topic_log", "base_log"));


        //
        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    try {
                        out.collect(JSON.parseObject(value));
                    } catch (JSONException e) {
                        System.out.println("无法解析为JSON格式脏数据：" + value);
                    }
                }
            }
        });

        //mid分组
        KeyedStream<JSONObject, String> keyedStream = jsonStream.keyBy(json -> json.getJSONObject("common").getString("mid"));

        //校验新老用户
        SingleOutputStreamOperator<JSONObject> jsonObjectIsnewStream = keyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {

            private ValueState<String> valueState;

            @Override

            public void open(Configuration parameters) throws Exception {
                valueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("valueString", String.class));
            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {
                String isNew = value.getJSONObject("common").getString("is_new");
                String stateDate = valueState.value();
                String ts = value.getString("ts");
                String dataDate = DateFormatUtil.toDate(Long.parseLong(ts));
                if ("1".equals(isNew)) {
                    if (stateDate == null || dataDate.equals(stateDate)) {
                        valueState.update(dataDate);
                    } else {
                        value.getJSONObject("common").put("is_new", "0");
                    }
                } else {
                    if (stateDate == null) {
                        valueState.update("1970-01-01");
                    }
                }
                return value;
            }
        });

        //主流：页面 侧输出流：启动、曝光、动作、错误
        OutputTag<String> startTag = new OutputTag<String>("start") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("display") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("action") {
        };
        OutputTag<String> errorTag = new OutputTag<String>("error") {
        };
        OutputTag<String> appVideoTag = new OutputTag<String>("video") {
        };

        SingleOutputStreamOperator<String> pageStream = jsonObjectIsnewStream.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                //尝试获取error信息
                String err = value.getString("err");
                if (err != null) {
                    ctx.output(errorTag, value.toJSONString());
                    value.remove("err");
                }


                //尝试获取start信息
                String start = value.getString("start");
                //尝试获取appVideo数据
                String appVideo = value.getString("appVideo");
                if (start != null) {
                    ctx.output(startTag, value.toJSONString());
                } else if (appVideo != null) {
                    ctx.output(appVideoTag,value.toJSONString());
                } else {
                    String pageId = value.getJSONObject("page").getString("page_id");
                    String ts = value.getString("ts");

                    //尝试获取曝光数据
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            display.put("page_id", pageId);
                            display.put("ts", ts);
                            ctx.output(displayTag, display.toJSONString());
                        }
                    }

                    //尝试获取动作数据
                    JSONArray actions = value.getJSONArray("actions");
                    if (actions != null) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            action.put("page_id", pageId);
                            ctx.output(actionTag, action.toJSONString());
                        }
                    }


                    //输出页面信息
                    value.remove("displays");
                    value.remove("actions");
                    out.collect(value.toJSONString());

                }


            }
        });

        DataStream<String> startOutput = pageStream.getSideOutput(startTag);
        DataStream<String> displayOutput = pageStream.getSideOutput(displayTag);
        DataStream<String> actionOutput = pageStream.getSideOutput(actionTag);
        DataStream<String> errorOutput = pageStream.getSideOutput(errorTag);
        DataStream<String> videoOutput = pageStream.getSideOutput(appVideoTag);

        startOutput.print("start>>");
        displayOutput.print("display>>");
        actionOutput.print("action>>");
        errorOutput.print("error>>");
        pageStream.print("page>>");
        videoOutput.print("video>>");

        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";
        String video_topic = "dwd_traffic_video_log";

        startOutput.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        pageStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        displayOutput.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionOutput.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errorOutput.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));
        videoOutput.addSink(MyKafkaUtil.getFlinkKafkaProducer(video_topic));

        env.execute("BaseLogApp");
    }
}
